package com.chiyoyo.rabbitmq.framework.consumer;

import cn.hutool.core.util.TypeUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.chiyoyo.rabbitmq.framework.annotation.RabbitMq;
import com.chiyoyo.rabbitmq.framework.annotation.RabbitMqRetry;
import com.chiyoyo.rabbitmq.framework.dto.RabbitMqModel;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;

import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;

/**
 * @author chil
 * RabbitMq消费器
 */
@Slf4j
public abstract class AbstractConsumer<T, R> extends MessageListenerAdapter {

    private final int retryNumber = 1;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        String correlationId = messageProperties.getHeader("spring_returned_message_correlation");
//        boolean acked = false;
        try {
            // 消费者内容
            String json = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("线程名:{},AbstractConsumer:消费者消息: {}", Thread.currentThread().getName(), json);
            RabbitMqModel<T> rabbitMqModel = new RabbitMqModel<>();
            Type typeArgument = TypeUtil.getTypeArgument(getTypeReference().getType(), 0);
            if ("java.lang.String".equals(typeArgument.getTypeName())) {
                rabbitMqModel.setBody((T) json);
            } else {
                rabbitMqModel = JSON.parseObject(json, getTypeReference());
                // 检查RabbitMqModel的body是否为空，若为空则使用json字符串解析为body。
                checkRabbitMqModel(rabbitMqModel, json);
            }
            // 判断是否重复消费
            if (checkMessageKey(messageProperties, rabbitMqModel)) {
                throw new RuntimeException(String.format("重复消费消息,correlationId为:%s", correlationId));
            }
            // 消费消息
            R result = this.handleMessage(rabbitMqModel.getBody(), message, channel);
            if (!isAutoAck()) {
                // 手动确认消息
                // ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
//                acked = true;
                channel.basicAck(deliveryTag, false);
            }
            // 保存消费成功消息
            saveLog(result, message, rabbitMqModel);
        } catch (Throwable e) {
            // 消费失败次数不等于空并且失败次数大于某个次数,不处理直接return,并记录到数据库
            log.error("AbstractConsumer:消费报错 异常为:", e);
            // 将消费失败的记录保存到数据库或者不处理也可以
//            this.saveFailMessage(message, e);
            // 保存重试失败次数达到retryNumber上线后拒绝此消息入队列并删除redis
//            saveFailNumber(messageProperties, channel, deliveryTag, correlationId, acked);
            throw e;
        } finally {
            // 删除判断重复消费Key
            deleteCheckMessageKey(messageProperties);
        }
    }

    /**
     * 检查RabbitMqModel的body是否为空，若为空则使用json字符串解析为body。
     *
     * @param rabbitMqModel 需要检查的RabbitMqModel对象
     * @param json          json字符串
     */
    private void checkRabbitMqModel(RabbitMqModel<T> rabbitMqModel, String json) {
        if (rabbitMqModel.getBody() == null || !"".equals(rabbitMqModel.getBody())) {
            // 获取泛型的实际类型
            Type type = TypeUtil.getTypeArgument(getTypeReference().getType(), 0);
            rabbitMqModel.setBody(JSON.parseObject(json, type));
        }
    }

    /**
     * 记录失败次数并决定是否拒绝此消息
     *
     * @param messageProperties 消息属性对象
     * @param channel           消息通道
     * @param deliveryTag       消息传递标签
     * @param correlationId     消息相关性ID
     * @param acked             消息是否被确认
     * @throws Exception 异常
     */
    public void saveFailNumber(MessageProperties messageProperties, Channel channel, long deliveryTag, String correlationId, boolean acked) throws Exception {
        Integer lock = messageProperties.getHeader("retryNumber");
        Integer actualLock = lock == null ? 1 : lock + 1;
        log.error("rabbitMQ 失败记录:消费者correlationId为:{},deliveryTag为:{},失败次数为:{}", correlationId, deliveryTag, actualLock);
        int retryNumber = getRetryNumber();
        if (retryNumber <= this.retryNumber || actualLock >= retryNumber) {
            if (retryNumber <= this.retryNumber) {
                log.error("rabbitMQ 失败记录:因记录不需要重试因此直接拒绝此消息,消费者correlationId为:{},消费者设置重试次数为:{}", correlationId, retryNumber);
            } else {
                log.error("rabbitMQ 失败记录:已满足重试次数,删除redis消息并且拒绝此消息,消费者correlationId为:{},重试次数为:{}", correlationId, actualLock);
            }
            if (!acked && channel.isOpen()) {
                channel.basicNack(messageProperties.getDeliveryTag(), false, false);
            }
        } else {
            log.error("rabbitMQ 失败记录:因记录重试次数还未达到重试上限，还将继续进行重试,消费者correlationId为:{},消费者设置重试次数为:{},现重试次数为:{}", correlationId, retryNumber, actualLock);
            messageProperties.setHeader("retryNumber", actualLock);
        }
    }

    /**
     * 获取重试次数，默认为2
     */
    public int getRetryNumber() {
        RabbitMqRetry rabbitMqRetry = getClass().getAnnotation(RabbitMqRetry.class);
        return rabbitMqRetry == null ? retryNumber : rabbitMqRetry.retryNumber();
    }

    /**
     * 获取是否是自动确认
     */
    public boolean isAutoAck() {
        RabbitMq rabbitMq = getClass().getAnnotation(RabbitMq.class);
        return rabbitMq != null && AcknowledgeMode.AUTO.toString().equalsIgnoreCase(rabbitMq.mode().toString());
    }

    /**
     * 消费方法
     *
     * @param body 请求数据
     */
    public abstract R handleMessage(T body, Message message, Channel channel) throws Exception;

    /**
     * 获取 TypeReference
     */
    public abstract TypeReference<RabbitMqModel<T>> getTypeReference();

    /**
     * 保存消费失败的消息
     *
     * @param message mq所包含的信息
     * @param e       异常
     */
    public void saveFailMessage(Message message, Throwable e) {
    }

    /**
     * 判断是否重复消费
     */
    public boolean checkMessageKey(MessageProperties messageProperties, RabbitMqModel<T> rabbitMqModel) {
        return false;
    }

    /**
     * 删除判断重复消费Key
     */
    public void deleteCheckMessageKey(MessageProperties messageProperties) {
    }

    /**
     * 保存消费成功消息
     */
    public abstract void saveLog(R result, Message message, RabbitMqModel<T> rabbitMqModel);

}
